package com.aotain.metis.bolt;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearch;
import com.aotain.apollo.IPSearcher;
import com.aotain.common.CommonFunction;
import com.aotain.mushroom.Slave;

/**
 * 
 * @ClassName: SourceFmtMonBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年11月30日 下午4:43:39 
 *
 */
public class SourceFmtMonBolt implements IRichBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 7816556443307732593L;

	Slave s = null;
	String sourceAreaCountry = null;
	String sourceAreaCity = null;
	String sourceAreaProvince = null;
	String sourceAreaName = null;
	String destAreaCountry = null;   
	String destAreaCity = null;
	String destAreaName = null;
	String destAreaProvince = null;
	String gisstr = null;
	String gisstrBase64 = null;
	String Sourcelon = null;
	String Sourcelat = null;
	String Destlon = null;
	String Destlat = null;
	String sourceAreaId = null;
	String sourceAreaProvinceId = null;
	String sourceAreaCityId = null;
	String destAreaId = null;
	String destAreaProvinceId = null;
	String destAreaCityId = null;
	String areaName_s = null;
	String areaName_d = null;
	String dip = null;
	String sip = null;
	String userAgent = "";
	String cookie = "";
	OutputCollector collector = null;
	HashMap<Long, IPDatabase> map = null;
	Long[] IPs = null;
	String dipL = null;
	String sipL = null;
	String dport = null;
	String sport = null;
	String filterIPs = null;
	String domainname = null;
	String url = null;
	String type = null;
	String accesstime = null;
	String idcno = null;
	String delaytime = null;
	long upStreamOctets = 0;
	long upStreamPacket = 0;
	long dnStreamOctets = 0;
	long dnStreamPacket = 0;
	Integer status ;
	String dsip = null;
	//	ApolloConfig ap = null;
//	IPSearch ipsearch = null;
//	IPSearcher ipsearcher = null;
	String logserver = null;
	String zooserver = null;
	String refer = "";
	String dbJson = "";
	HashSet<String> IPSet = new HashSet<String>();
	HashSet<String> WhiteIPSet = new HashSet<String>();

	public SourceFmtMonBolt(){}

	public SourceFmtMonBolt(String filterIPs) {
		this.filterIPs = filterIPs;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","dport","sip","sport","gisstr","domainname","url","type","delaytime","idcno","upstreamoctets","upstreampacket","status","useragent","cookie","dnstreamoctets","dnstreampacket","refer"));
	}   

	@Override
	public Map<String, Object> getComponentConfiguration() {
		//		Map<String,Object> conf = new HashMap<String,Object>();
		//		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,30*60);
		//		return conf;  
		return null;
	}

	@Override
	public void execute(Tuple tuple) {
		try {
			//			if(TupleHelpers.isTickTuple(tuple)) {
			//				emitCountingData();  	
			//			} else {
			countInLocal(tuple); 
			collector.ack(tuple);
			//			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error("SourceFmtMonBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void countInLocal(Tuple tuple) {
		// TODO Auto-generated method stub
		String logString = tuple.getString(0);
		if(logString != null && !"".equals(logString)) {
			String[] splits = logString.split("\\|",-1);
			if(splits.length <= 1)
				return;
			String userAccount = splits[1];
			String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
			Pattern pattern = Pattern.compile(ipRex);   
			Matcher matcher = pattern.matcher(userAccount);  
			if(matcher.matches()) {
				if("1".equals(splits[22]) && "6".equals(splits[7]) && splits.length > 42) {//http
					dipL = splits[9];
					dport = splits[10];
					sipL = splits[12];
					sport = splits[13];
					domainname = splits[32];
					url = domainname + "/" + splits[28];
					type = splits[7]; 
					accesstime= splits[14];
					idcno = "911";
					dnStreamOctets = Long.valueOf(splits[17]);
					upStreamOctets = Long.valueOf(splits[18]);
					dnStreamPacket = Long.valueOf(splits[19]);
					upStreamPacket = Long.valueOf(splits[20]);
					status = Integer.valueOf(splits[23]);
					userAgent = splits[26];
					cookie = splits[42];
					refer = splits[37];
				}else{//非Http
					dipL = splits[9];
					dport = splits[10];
					sipL = splits[12];
					sport = splits[13];
					domainname = "";
					url = "";
					type = splits[7]; 
					accesstime= splits[14];
					idcno = "911";
					dnStreamOctets = Long.valueOf(splits[17]);
					upStreamOctets = Long.valueOf(splits[18]);
					dnStreamPacket = Long.valueOf(splits[19]);
					upStreamPacket = Long.valueOf(splits[20]);
					status = Integer.valueOf(splits[23]);
				} /*else {
					return;
				}*/

				delaytime = 0 + "";
				long currenttime = System.currentTimeMillis();
				if(accesstime != null && !"".equals(accesstime)) {
					delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
				}

				if(!"".equals(dipL) && !"".equals(dport) && !"".equals(sipL) && !"".equals(sport) ) {
					dip = CommonFunction.int2ip(Long.parseLong(dipL));
					sip = CommonFunction.int2ip(Long.parseLong(sipL));
					//白名单过滤
					//					dsip = dip.trim() + "#" + sip.trim();
					//					if("121.10.255.20#192.168.5.35".equals(dsip) || "121.10.255.20#183.62.140.142".equals(dsip))
					//					if(!WhiteIPSet.contains(dsip)) {
					
					long sipLong = Long.parseLong(sipL);
					if((sipLong >= 167772160 && sipLong <= 184549375) 
						|| (sipLong >= 2886729728L && sipLong <= 2887778303L)
						|| (sipLong >= 3232235520L &&sipLong <= 3232301055L)) {
						
						IPSearcher ipsearcher = IPSearcher.getInstance(dbJson);
						IPDatabase SourceArea = ipsearcher.SearchIPDB(sip);
						IPDatabase DestArea = ipsearcher.SearchIPDB(dip);

						if(DestArea != null) {
							destAreaCountry = DestArea.getCountryName();
							destAreaCity = DestArea.getCityName();
							//						sourceAreaProvince = SourceArea.getProviceName();
							if(destAreaCity != null) {
								destAreaName  = destAreaCity;
								destAreaId = DestArea.getCityID()+"";
							} else {
								destAreaName = destAreaCountry;
								destAreaId = DestArea.getCountryID()+"";
							}
							destAreaCityId = sourceAreaId;
							Destlon = String.valueOf(DestArea.getLon());
							Destlat = String.valueOf(DestArea.getLat());

							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
								destAreaId = DestArea.getProviceID()+"";
								areaName_d = DestArea.getProviceName().replace("省", "");
								areaName_d = areaName_d.replace("市", "");
							} else {
								areaName_d = DestArea.getCountryName();
								destAreaId = DestArea.getCountryID()+"";
							}
							destAreaProvinceId = destAreaId;
						}   

						if(SourceArea != null) {    
							sourceAreaCountry = SourceArea.getCountryName();
							sourceAreaCity = SourceArea.getCityName();
							if(sourceAreaCity != null) {
								sourceAreaName = sourceAreaCity;
							} else {   
								sourceAreaName = sourceAreaCountry;
							}
							Sourcelon = String.valueOf(SourceArea.getLon());//经度
							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							collector.emit(new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,dnStreamOctets,dnStreamPacket,refer));
							collector.ack(tuple);
						}
						//					} else {
						//						System.out.println("############WhiteIPSet-------dsip" + dsip);
						//					}
					}	
				}
			}
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		if(!logserver.isEmpty())
		{
			Slave s = new Slave(zooserver);

			s.SendAlarmLog("ERROR", "SourceFmtMonBolt Shutdown");
		}
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.collector = collector;

		dbJson = stormConf.get("dbjson").toString();
		logserver = stormConf.get("mushroomserver").toString();
		zooserver = stormConf.get("zkConn").toString();
//		ipsearch = new IPSearch(dbJson);
//		ipsearcher = IPSearcher.getInstance(dbJson);
		//ip过滤
		//		s = new Slave(zooserver);
		//		IPSet = s.GetIPFilterForLog();
		//		WhiteIPSet = s.GetIPWhiteList();
		//		for(String s :WhiteIPSet) {
		//			System.out.println("############WhiteIPSet=" + s);
		//		}
	}

	public void emitCountingData() {

		//		if(s.GetIPFilterForLog() != null && s.GetIPFilterForLog().size() > 0) {
		//			IPSet = s.GetIPFilterForLog();
		//			WhiteIPSet = s.GetIPWhiteList();
		//		}

	}

}
